package cn.itcast.kafka;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

//模拟消费kafka数据
public class FlinkKafkaConsumer {
    public static void main(String[] args) throws Exception {
        //
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(1000*30);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        Properties prop = new Properties();
        prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node01:9092,node02:9092,node03:9092");
        prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"group_01");
        prop.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");//消费策略
        FlinkKafkaConsumer011<String> consumer011 = new FlinkKafkaConsumer011<>(
                "vehicledata-dev",
                new SimpleStringSchema(), prop);

        DataStreamSource<String> streamSource = env.addSource(consumer011);

        streamSource.printToErr();

        env.execute();


    }
}
